package simpledb;

import java.util.*;
import java.io.*;

/**
 * HeapPage stores pageList of HeapFiles and implements the Page interface that
 * is used by BufferPool.
 *
 * @see HeapFile
 * @see BufferPool
 */
public class HeapPage implements Page {


    boolean dirty = false ;
    TransactionId transactionId ;

    HeapPageId pid;
    TupleDesc td;
    byte header[];
    Tuple tuples[];
    int numSlots;

    byte[] oldData;

    public byte[] getHeader() {
        return header;
    }

    public void setHeader(byte[] header) {
        this.header = header;
    }

    /**
     * Create a HeapPage from a set of bytes of data read from disk.
     * The format of a HeapPage is a set of header bytes indicating
     * the slots of the page that are in use, some number of tuple slots.
     *  Specifically, the number of tuples is equal to: <p>
     *          floor((BufferPool.PAGE_SIZE*8) / (tuple size * 8 + 1))
     * <p> where tuple size is the size of tuples in this
     * database table, which can be determined via {@link Catalog#getTupleDesc}.
     * The number of 8-bit header words is equal to:
     * <p>
     *      ceiling(no. tuple slots / 8)
     * <p>
     * @see Database#getCatalog
     * @see Catalog#getTupleDesc
     * @see BufferPool#PAGE_SIZE
     */
    public HeapPage( HeapPageId id, byte[] data) throws IOException {

        this.pid = id ;
        this.td = Database.getCatalog().getTupleDesc( id.getTableId() );
        this.numSlots = getNumTuples();

        System.out.println( "===this.numSlots ::" + this.numSlots );
        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(data));

        // allocate and read the header slots of this page
        header = new byte[getHeaderSize()];
        for (int i=0; i<header.length; i++)
            header[i] = dis.readByte();

        try{
            // allocate and read the actual records of this page
//            int tupleByteLength = data.length - header.length ;
//            int tupleSize  = tupleByteLength /td.getSize()  ;
//
//            tuples = new Tuple[tupleSize];
//            for (int i=0; i<tuples.length; i++){
//                tuples[i] = readNextTuple(dis,i);
//            }
            tuples = new Tuple[numSlots];
            for (int i=0; i<tuples.length; i++){
                tuples[i] = readNextTuple(dis,i);
            }

        }catch(NoSuchElementException e){
            e.printStackTrace();
        }
        dis.close();

        setBeforeImage();
    }

    /** Retrieve the number of tuples on this page.
        @return the number of tuples on this page
    */
    private int getNumTuples() {        
        // some code goes here

        int tupleSize = 0 ;

        for (int i = 0; i < this.td.numFields() ; i++) {
            tupleSize += this.td.getType(i).getLen();
        }

//        System.out.println( "padding..." + (npagebytes - (recordcount * nrecbytes + nheaderbytes)) );

        int nrecords = ( BufferPool.PAGE_SIZE * 8) /  (tupleSize * 8 + 1);  //floor comes for free

        return nrecords ;
    }

    /**
     * Computes the number of bytes in the header of a page in a HeapFile with each tuple occupying tupleSize bytes
     * @return the number of bytes in the header of a page in a HeapFile with each tuple occupying tupleSize bytes
     */
    private int getHeaderSize() {

        // some code goes here
        int nrecbytes = 0;
        for (int i = 0; i < this.td.numFields()  ; i++) {
            nrecbytes +=  this.td.getType(i).getLen();
        }

//        int nrecords = ( npagebytes * 8) / (tupleSize* 8 + 1) ;
        int nrecords = ( BufferPool.PAGE_SIZE * 8) /  (nrecbytes * 8 + 1);  //floor comes for free

        int nheaderbytes = (nrecords / 8);
        if (nheaderbytes * 8 < nrecords)
            nheaderbytes++;  //ceiling

//        int nheaderbits = nheaderbytes * 8;

//        double headerBytes = Math.ceil( tupsPerPage/8 );

        return  nheaderbytes ;
    }
    
    /** Return a view of this page before it was modified
        -- used by recovery */
    public HeapPage getBeforeImage(){
        try {
            return new HeapPage(pid,oldData);
        } catch (IOException e) {
            e.printStackTrace();
            //should never happen -- we parsed it OK before!
            System.exit(1);
        }
        return null;
    }
    
    public void setBeforeImage() {
        oldData = getPageData().clone();
    }

    /**
     * @return the PageId associated with this page.
     */
    public HeapPageId getId() {

        return this.pid ;
    // some code goes here
//    throw new UnsupportedOperationException("implement this");
    }

    /**
     * Suck up tuples from the source file.
     */
    private Tuple readNextTuple(DataInputStream dis, int slotId) throws NoSuchElementException {
        // if associated bit is not set, read forward to the next tuple, and
        // return null.
        if (!getSlot(slotId)) {
            for (int i=0; i<td.getSize(); i++) {
                try {
                    dis.readByte();
                } catch (IOException e) {
                    throw new NoSuchElementException("error reading empty tuple");
                }
            }
            return null;
        }

        // read fields in the tuple
        Tuple t = new Tuple(td);
        RecordId rid = new RecordId(pid, slotId);
        t.setRecordId(rid);
        try {
            for (int j=0; j<td.numFields(); j++) {
                Field f = td.getType(j).parse(dis);
                t.setField(j, f);
            }
        } catch (java.text.ParseException e) {
            e.printStackTrace();
            throw new NoSuchElementException("parsing error!");
        }

        return t;
    }

    /**
     * Generates a byte array representing the contents of this page.
     * Used to serialize this page to disk.
     * <p>
     * The invariant here is that it should be possible to pass the byte
     * array generated by getPageData to the HeapPage constructor and
     * have it produce an identical HeapPage object.
     *
     * @see #HeapPage
     * @return A byte array correspond to the bytes of this page.
     */
    public byte[] getPageData() {
        int len = BufferPool.PAGE_SIZE;
        ByteArrayOutputStream baos = new ByteArrayOutputStream(len);
        DataOutputStream dos = new DataOutputStream(baos);

        // create the header of the page
        for (int i=0; i<header.length; i++) {
            try {
                dos.writeByte(header[i]);
            } catch (IOException e) {
                // this really shouldn't happen
                e.printStackTrace();
            }
        }

        // create the tuples
        for (int i=0; i<tuples.length; i++) {

            // empty slot
            if (!getSlot(i)) {
                for (int j=0; j<td.getSize(); j++) {
                    try {
                        dos.writeByte(0);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }

                }
                continue;
            }

            // non-empty slot
            for (int j=0; j<td.numFields(); j++) {
                Field f = tuples[i].getField(j);
                try {
                    f.serialize(dos);
                
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        // padding
        int zerolen = BufferPool.PAGE_SIZE - (header.length + td.getSize() * tuples.length); //- numSlots * td.getSize();
        byte[] zeroes = new byte[zerolen];
        try {
            dos.write(zeroes, 0, zerolen);
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            dos.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }

        return baos.toByteArray();
    }

    /**
     * Static method to generate a byte array corresponding to an empty
     * HeapPage.
     * Used to add new, empty pageList to the file. Passing the results of
     * this method to the HeapPage constructor will create a HeapPage with
     * no valid tuples in it.
     *
     * @return The returned ByteArray.
     */
    public static byte[] createEmptyPageData() {
        int len = BufferPool.PAGE_SIZE;
        return new byte[len]; //all 0
    }

    /**
     * Delete the specified tuple from the page;  the tuple should be updated to reflect
     *   that it is no longer stored on any page.
     * @throws DbException if this tuple is not on this page, or tuple slot is
     *         already empty.
     * @param t The tuple to delete
     */
    public void deleteTuple(Tuple t) throws DbException {
        // some code goes here
        // not necessary for lab1

        RecordId recordId = t.getRecordId();
        if( recordId.getPageId().equals(pid) ){

            int  emptySlotsIndex = recordId.tupleno() ;
            if( getSlot( emptySlotsIndex ) ){
                setSlot( emptySlotsIndex , false );
                tuples[emptySlotsIndex] = null ;
            }else{
                throw new DbException("not delete success..");
            }
        }else{
            throw new DbException("not delete success..");
        }

    }

    /**
     * Adds the specified tuple to the page;  the tuple should be updated to reflect
     *  that it is now stored on this page.
     * @throws DbException if the page is full (no empty slots) or tupledesc
     *         is mismatch.
     * @param t The tuple to add.
     */
    public void addTuple(Tuple t) throws DbException {
        // some code goes here
        // not necessary for lab1

        int emptySlotsIndex = -1 ;
        for (int i = 0; i < header.length*8 ; i++) {
            if( !getSlot(i) ){
//                System.out.println(" is empty slot .. i==" + i );
                emptySlotsIndex = i ;
                break;
            }
        }

        if( emptySlotsIndex != -1 ){
            setSlot( emptySlotsIndex , true );
            RecordId recordId = new RecordId( this.pid , emptySlotsIndex );
            t.setRecordId( recordId );

            tuples[emptySlotsIndex] = t ;
        }else{
            throw new DbException(" the page is full (no empty slots)");
        }

    }


    /**
     * Marks this page as dirty/not dirty and record that transaction
     * that did the dirtying
     */
    public void markDirty(boolean dirty, TransactionId tid) {
        // some code goes here
	// not necessary for lab1

        this.dirty = dirty  ;
        this.transactionId = tid ;
    }

    /**
     * Returns the number of empty slots on this page.
     */
    public int getNumEmptySlots() {
        // some code goes here

        int count = 0;
        for (int i = 0; i < header.length*8 ; i++) {
            if( !getSlot(i) ){
//                System.out.println(" is empty slot .. i==" + i );
                count++;
            }
        }
        return count ;
    }

    /**
     * Returns the tid of the transaction that last dirtied this page, or null if the page is not dirty
     */
    public TransactionId isDirty() {
        // some code goes here
        // Not necessary for lab1

        if( this.dirty ){
            return transactionId ;
        }

        return null;
    }

    /**
     * Returns true if associated slot on this page is filled.
     */
    public boolean getSlot(int i) {

//        System.out.println( header.length );

        //63*8=504
        int index = i / 8;

//        System.out.println( ":::: getSlot 111 index ::: " + index + " i == " + i  );

        byte slotByte = header[index];

        int result = getByte( slotByte , i%8 ) ;
        if( result == 1){
            return true ;
        }else{
            return false ;
        }
    }


    private int getByte( byte b,int bit) {
        if( bit < 0 || bit > 7)
            return 0;
        return ( b & (0b1 << bit)) > 0 ? 1: 0;
    }

    /**
     * Abstraction to fill or clear a slot on this page.
     */
    private void setSlot(int i , boolean value) {
        // some code goes here
        // not necessary for lab1

        //63*8=504
        int index = i / 8;

//        System.out.println( ":::: getSlot 111 index ::: " + index + " i == " + i  );

        byte slotByte = header[index];
        int pos = i%8 ;

//      如果是设置为0
        if( !value ){
            slotByte = (byte)(slotByte & ~(1 << pos)) ;
        }else{
//      如果是设置为1
            slotByte = (byte)(slotByte | (1 << pos)) ;
        }
        header[index] = slotByte ;

//        int result = getByte( slotByte ,  ) ;
//        if( result == 1){
//            return true ;
//        }else{
//            return false ;
//        }

    }

    /**
     * @return an iterator over all tuples on this page (calling remove on this iterator throws an UnsupportedOperationException)
     * (note that this iterator shouldn't return tuples in empty slots!)
     */
    public Iterator<Tuple> iterator() {
        // some code goes here

        List<Tuple> list = new ArrayList();
        for (int i = 0; i < this.tuples.length ; i++) {
//            if( i == 20 ){
//                System.out.println( " aa= " );
//            }
//            System.out.println( " aa= "  + i );

            if( tuples[i] != null)
                list.add( tuples[i] );
        }

        return list.iterator() ;
    }

}

